package cn.edu.ccnu.nlp;

import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.util.Bytes;

public class DataImportUpdate
{
	public static Configuration configuration;
	public static HTable news_table;
	public static HTable news_index;
	public static HTable stat_table;
	static
	{
		configuration = HBaseConfiguration.create();
		//configuration.set("hbase.master", "pc1:60000");
		//configuration.set("hbase.zookeeper.property.clientPort", "2222");
		//configuration.set("hbase.zookeeper.quorum", "pc1,pc2,pc3");
	try
	{
		news_table = new HTable(configuration, "news");
		stat_table = new HTable(configuration, "news_stat");
		news_index = new HTable(configuration, "news_index");
	}
	catch (IOException e)
	{
		e.printStackTrace();
	}
	}
	
	public Connection getConnection()
	{
		String url = "jdbc:mysql://202.114.40.175:3306/corpus?useUnicode=true&characterEncoding=UTF-8";
		String name = "root";
		String psw = "ccnunlp2010@nlp.com";
		Connection conn = null;
	try
	{
		Class.forName("com.mysql.jdbc.Driver");
		conn = DriverManager.getConnection(url, name, psw);
		System.out.println("connect successful !");
	}
	catch (Exception e)
	{
		e.printStackTrace();
	}
	return conn;
	}
	
	/**
	* 将mysql的news_stat表迁移到hbase
	* @throws SQLException
	*/
	public void importNewsStat() throws SQLException
	{
		Connection conn = getConnection();
		Statement stmt = null;
		ResultSet rs = null;
		try
		{
			stmt = conn.createStatement();
	
			String sql = "SELECT count(*)  FROM news_stat";
	
			rs = stmt.executeQuery(sql);
			rs.next();
			int count = rs.getInt(1);
			System.out.println("总记录数为：" + count);
	
			for (int i = 0; i <= count / 5000; i++)
			{
				stat_table.setAutoFlush(false);
	
				sql = "SELECT *  FROM news_stat limit " + i * 5000 + "," + 5000;
				rs = stmt.executeQuery(sql);
	
				while (rs.next())
					insertNewsStat( rs.getString("date"), rs.getString("num"));
	
				stat_table.flushCommits();
				System.out.println("第" + (i + 1) * 5000 + "条记录导入成功！");
			}
		}
		catch (Exception e)
		{
			e.printStackTrace(System.out);
		}
		finally
		{
			rs.close();
			stmt.close();
			conn.close();
		}
	}
	
	public void importNews() throws SQLException
	{
		Connection conn = getConnection();;
		Statement stmt = null;
		ResultSet rs = null;
		String sql = null;
		try
		{
			stmt = conn.createStatement();
			sql = "SELECT count(*)  FROM news";
			
			rs = stmt.executeQuery(sql);
			rs.next();
			int count = rs.getInt(1);
			System.out.println("news_table.isAutoFlush()  : " + news_table.isAutoFlush());
			String author;
			int itecount = count / 5000;
			
			for (int i = 0; i <= itecount; i++)
			{
				System.out.println(i + "th of " + itecount);
				sql = "SELECT *  FROM news limit " + i * 5000 + "," + 5000;
				
				rs = stmt.executeQuery(sql);
				
				news_table.setAutoFlush(false);
				
				while (rs.next())
				{
					if (rs.getString("author") == null)
						author = "";
					else
						author = rs.getString("author");
				
					insertNews( rs.getString("url"), rs.getString("title"), author, rs.getString("date") + " " + rs.getString("time"), rs.getString("source"), rs.getString("content"));
				}
				news_table.flushCommits();
			}
		}
		catch (Exception e)
		{
			e.printStackTrace(System.out);
		}
		finally
		{
			rs.close();
			stmt.close();
			conn.close();
		}
	}
	
	//   sql 那个表里的字段是 'news','url','title','author','date','time',source','content
	    //   而你的字段是        url    title  author    date         source   text
	/**
	* @param url  
	* @param title
	* @param author
	* @param date
	* @param source
	* @param html	//为空
	* @param text    -> 换名为 content
	* @param seg	//为空
	*/
	public void insertNews(String url, String title, String author, String date, String source, String text) throws IOException
	{
		Put put = new Put(Bytes.toBytes(url));
		put.add(Bytes.toBytes("title"), null, title.getBytes());
		put.add(Bytes.toBytes("author"), null, author.getBytes());
		put.add(Bytes.toBytes("date"), null, date.getBytes());
		put.add(Bytes.toBytes("source"), null, source.getBytes());
		put.add(Bytes.toBytes("text"), null, text.getBytes());
		news_table.put(put);
	}
	
	/**
	 * 导入mysql数据到模拟索引的新闻表
	 * @throws SQLException
	 */
	/*public void importNewsIndex() throws SQLException
	{
		Connection conn = getConnection();;
		Statement stmt = null;
		ResultSet rs = null;
		String sql = null;
		try
		{
			stmt = conn.createStatement();
			sql = "SELECT count(*)  FROM news";
			
			rs = stmt.executeQuery(sql);
			rs.next();
			int count = rs.getInt(1);
			System.out.println("news_table.isAutoFlush()  : " + news_table.isAutoFlush());
			
			int itecount = count / 5000;
			
			for (int i = 0; i <= 5; i++)
			{
				System.out.println(i + "th of " + itecount);
				sql = "SELECT *  FROM news limit " + i * 5000 + "," + 5000;
				
				rs = stmt.executeQuery(sql);
				rs = stmt.executeQuery(sql);
				news_table.setAutoFlush(false);
				while (rs.next())
				{	
						String time = rs.getString("time");
						if(time == null)
							time = "00:00:00";
						insertNewsIndex( rs.getString("url"), rs.getString("title"), rs.getString("date") , time);
				}
				news_table.flushCommits();
			}
		}
		catch (Exception e)
		{
			e.printStackTrace(System.out);
		}
		finally
		{
			rs.close();
			stmt.close();
			conn.close();
		}
	}
	*/
	/**
	 * 循环读取news表中的数据，建立news的模拟索引表news_index
	 */
	public void importNewsIndex(){
		
		int pageSize = 1000;               //每次读取1000条记录
		byte[] lastRow = null;
		int totalRows = 0;
		ResultScanner rs = null;
		int i = 0;
		
		Filter filter = new PageFilter(pageSize);
		while(true){
			System.out.println("正在读取第"+((i*pageSize)+1)+"--"+((i+1)*pageSize)+"记录.....");
			Scan scan = new Scan();
			scan.setFilter(filter);
			if(lastRow != null){
				byte[] startRow = Bytes.add(lastRow, lastRow);
				scan.setStartRow(startRow);
			}
			
			int localRows = 0;
			
			try{
				rs = news_table.getScanner(scan);
				news_index.isAutoFlush();
				Result result;
				while((result = rs.next()) != null){
					for(KeyValue keyValue:result.raw()){
						KeyValue title = result.getColumnLatest("title".getBytes(), null);
						KeyValue date = result.getColumnLatest("date".getBytes(), null);     //获取date(包含了日期和时间)，根据分隔符切割需注意分隔符
						String date1 = new String(date.getValue());
						String time = date1.split(" ")[1];
						//System.out.println(date1);
						if(time.equals("null"))
							time = "00:00:00";
						insertNewsIndex( new String(result.getRow()),new String(title.getValue()),date1 ,time);  //插入到news_index表中
						totalRows++;
					}
					lastRow = result.getRow();
					localRows++;
				}
				news_index.flushCommits();
			}catch(Exception e){
				e.getStackTrace();
			}
			rs.close();
			if(localRows == 0) break;
			i++;
		}
		System.out.println("共插入"+totalRows+"条数据，news_index表创建成功！");
	}
	/**
	 * 插入一条记录到hbase中
	 * @param url
	 * @param title
	 * @param date
	 * @param time
	 * @throws IOException
	 */
	public void insertNewsIndex(String url, String title,String date,String time) throws IOException
	{
		Put put = new Put(Bytes.toBytes(date));
		put.add(Bytes.toBytes("news"), url.getBytes(), (time+","+title).getBytes());
		news_index.put(put);
	}
	/**
	* 向news_stat表中插入数据
	*
	* @param tablename
	* @param date
	* @param num
	* @throws IOException
	*/
	public void insertNewsStat(String date, String num) throws IOException
	{
		Put put = new Put(Bytes.toBytes(date));
		put.add(Bytes.toBytes("num"), null, num.getBytes());
		stat_table.put(put);
	}
	
	public static void main(String[] args) throws SQLException
	{
		DataImportUpdate di = new DataImportUpdate();
	//	di.importNewsStat();
	//	di.importNews();
	//	di.importTestNews();
	//di.importNewsStat();
		di.importNewsIndex();
	}
}